/*
 * Copyright 2022 The Open Islands Authors. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.openislands.oi.scheduling;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.openislands.oi.util.CommonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

public class DAGOperator {
    private static Logger logger = LoggerFactory.getLogger(DAGOperator.class);
    private String name;                                            // 组件名字 比如reader_0
    private String module;                                          // 组件类型 比如"Reader"
    private Map<String, Object> param;                              // 组件参数 比如lr组件参数
    private Map<String, Object> resource;                           // 组件资源
    private DAGOperatorInput input;                                 // 组件多输入
    private DAGOperatorOutput output;                               // 组件多输出

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getModule() {
        return module;
    }

    public void setModule(String module) {
        this.module = module;
    }

    public Map<String, Object> getParam() {
        return param;
    }

    public void setParam(Map<String, Object> param) {
        this.param = param;
    }

    public Map<String, Object> getResource() {
        return resource;
    }

    public void setResource(Map<String, Object> resource) {
        this.resource = resource;
    }

    public DAGOperatorInput getInput() {
        return input;
    }

    public void setInput(DAGOperatorInput input) {
        this.input = input;
    }

    public DAGOperatorOutput getOutput() {
        return output;
    }

    public void setOutput(DAGOperatorOutput output) {
        this.output = output;
    }

    // 合并两个算走
    public static DAGOperator merge(DAGOperator base, DAGOperator ow) throws JsonProcessingException, DAGException {
        // 判断name，module是否匹配
        if (base.getName().equals(ow.getName()) == false) {
            DAGErrorEnums error = DAGErrorEnums.OVERWRITE_OPERATOR_NAME_NOT_MATCH;
            logger.error("{} and {}, error={}", base.getName(), ow.getName(), error.getMessage());
            throw new DAGException(error);
        }
        if (base.getModule().equals(ow.getModule()) == false) {
            DAGErrorEnums error = DAGErrorEnums.OVERWRITE_OPERATOR_MODULE_NOT_MATCH;
            logger.error("{} and {}, error={}", base.getName(), ow.getName(), error.getMessage());
            throw new DAGException(error);
        }
        // 判断input是否匹配
        DAGOperatorInput baseInput = base.getInput();
        DAGOperatorInput owInput = ow.getInput();
        boolean match = DAGOperatorInput.equals(baseInput, owInput);
        if (match == false) {
            DAGErrorEnums error = DAGErrorEnums.OVERWRITE_OPERATOR_INPUT_NOT_MATCH;
            logger.error("{} and {}, error={}", base.getName(), ow.getName(), error.getMessage());
            throw new DAGException(error);
        }
        // 判断output是否匹配
        DAGOperatorOutput baseOutput = base.getOutput();
        DAGOperatorOutput owOutput = ow.getOutput();
        match = DAGOperatorOutput.equals(baseOutput, owOutput);
        if (match == false) {
            DAGErrorEnums error = DAGErrorEnums.OVERWRITE_OPERATOR_OUTPUT_NOT_MATCH;
            logger.error("{} and {}, error={}", base.getName(), ow.getName(), error.getMessage());
            throw new DAGException(error);
        }

        // 构造新算子；更新param和resource部分
        DAGOperator ret = DAGOperator.fromJson(base.toJson());
        // 更新param部分
        Map<String, Object> retParam = ret.getParam();
        Map<String, Object> owParam = ow.getParam();
        Map<String, Object> mergeParam = CommonUtils.merge(retParam, owParam);
        ret.setParam(mergeParam);
        // 更新resource部分
        Map<String, Object> retResource = ret.getResource();
        Map<String, Object> owResource = ow.getResource();
        Map<String, Object> mergeResource = CommonUtils.merge(retResource, owResource);
        ret.setResource(mergeResource);
        return ret;
    }

    // json字符串转换为结构体
    public static DAGOperator fromJson(String json) throws JsonProcessingException {
        ObjectMapper mapper = new ObjectMapper();
        DAGOperator op = mapper.readValue(json, DAGOperator.class);
        return op;
    }

    // 结构体转换为json字符串
    public String toJson() throws JsonProcessingException {
        Map<String, Object> map = new HashMap<>();
        map.put("name", name);
        map.put("module", module);
        map.put("param", param);
        map.put("resource", resource);
        map.put("input", input);
        map.put("output", output);
        ObjectMapper mapper = new ObjectMapper();
        String json = mapper.writeValueAsString(map);
        return json;
    }

}

